package com.xrui.hbasecmd.config;

import com.xrui.hbase.security.Kerboros;
import com.xrui.hbase.util.Resources;

import com.xrui.hbasecmd.actions.Action;
import com.xrui.hbasecmd.actions.RunnableAction;
import com.xrui.hbasecmd.hbase.ConnectionManager;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.net.MalformedURLException;
import java.net.URL;

/**
 * This class represents a data to be used to connect to the hbase and zookeeper nodes.
 */
public class ConnectionDetails implements Serializable {

    //region Constants
    private final static Logger LOG = LoggerFactory.getLogger(ConnectionDetails.class);
    private static final long serialVersionUID = -5808921673890223877L;
    //endregion

    //region Variables
    private String configurationPath;
    private ServerDetails zookeeper;
    private Configuration configuration;
    //endregion

    //region Constructor
    public ConnectionDetails(String name) {
        this.configurationPath = GlobalConfig.instance().getClusterConfigurationFolder() + '/' + name;
        this.configuration = createConfig();
        this.zookeeper = new ServerDetails(
            configuration.get(HConstants.ZOOKEEPER_QUORUM, "localhost"),
            configuration.get(HConstants.ZOOKEEPER_CLIENT_PORT, "2181")
        );
    }

    public ConnectionDetails() {
        this.configuration = HBaseConfiguration.create();
        this.zookeeper = new ServerDetails(
            configuration.get(HConstants.ZOOKEEPER_QUORUM, "localhost"),
            configuration.get(HConstants.ZOOKEEPER_CLIENT_PORT, "2181")
        );
    }
    //endregion

    //region Public Properties
    public ServerDetails getZookeeper() {
        return this.zookeeper;
    }
    //endregion

    //region Public Methods
    public boolean canConnect() {
        String operation = String.format("Testing connection to zookeeper %s", zookeeper.getHost());
        LOG.debug(operation + "...");
        Boolean result = RunnableAction.runAndWait(
            operation, new Action<Boolean>() {
                @Override
                public Boolean run() throws IOException {
                    ConnectionManager.get().create(ConnectionDetails.this);
                    LOG.debug("Connection available to zookeeper {}...", zookeeper.getHost());
                    return true;
                }

                @Override
                public void onError(Exception ex) {
                    LOG.warn(String.format("Connection not available to zookeeper %s", zookeeper.getHost()), ex);
                }
            }, GlobalConfig.instance().getConnectionCheckTimeout());

        return result != null ? result : false;
    }

    public Configuration createConfig() {
        if (configuration != null) {
            return configuration;
        }
        Configuration config = new Configuration();
        addResource(config, "hbase-site.xml");
        addResource(config, "core-site.xml");
        addResource(config, "hdfs-site.xml");

        String krbConfigKey = "java.security.krb5.conf";
        if (System.getProperty(krbConfigKey) == null) {
            File krbConfig = new File(configurationPath + '/' + "krb5.ini");
            if (krbConfig.isFile() && krbConfig.exists()) {
                System.setProperty(krbConfigKey, krbConfig.getAbsolutePath());
            } else {
                URL url = Resources.getResource("krb5.ini");
                if(null != url && url.getFile() != null) {
                    System.setProperty(krbConfigKey, url.getFile());
                }
            }
        }

        File keyTab = new File(configurationPath + '/' + "hbase.keytab");
        if (keyTab.isFile() && keyTab.exists()) {
            System.setProperty(Kerboros.KEY_TAB_DEFINITION, keyTab.getAbsolutePath());
            String principal = config.get("hbase.thrift.kerberos.principal");
            if(principal == null) {
                principal = config.get(Kerboros.PRINCIPAL_DEFINITION);
            }
            if (principal != null) {
                System.setProperty(Kerboros.PRINCIPAL_DEFINITION, principal);
            }
        }

        config.set("hbase.client.retries.number", "3");
        config.set("hbase.rpc.timeout", "900000");

        return config;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof ConnectionDetails) {
            ConnectionDetails details = (ConnectionDetails) obj;
            return this.zookeeper.equals(details.zookeeper);
        }
        return false;
    }

    @Override
    public int hashCode() {
        return this.zookeeper.hashCode();
    }
    //endregion

    //region Private Methods
    private void addResource(Configuration configuration, String name) {
        File file = new File(configurationPath + '/' + name);
        if (file.exists() && file.isFile()) {
            try {
                configuration.addResource(file.toURI().toURL());
            } catch (MalformedURLException e) {
                LOG.error("Failed to load resource {}", file.getAbsolutePath());
            }

        }
    }
    //endregion
}
